package com.fanshuai.spring.boot.interceptor;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;


import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.UUID;

@Slf4j
public class MyProducerInterceptor implements ProducerInterceptor<String, String> {
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        String traceId = UUID.randomUUID().toString();
        record.headers().add("traceId", traceId.getBytes(StandardCharsets.UTF_8));
        log.info("send kafka messages, record={}", record);

        return record;
    }

    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        log.info("producer acked, metadata={}", metadata);
    }

    public void close() {

    }

    public void configure(Map<String, ?> configs) {

    }
}
